package org.deng.learn.es.rest;

import org.apache.http.HttpHost;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.open.OpenIndexRequest;
import org.elasticsearch.action.admin.indices.open.OpenIndexResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.support.ActiveShardCount;
import org.elasticsearch.action.support.IndicesOptions;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.client.HttpAsyncResponseConsumerFactory;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.GetSourceRequest;
import org.elasticsearch.client.core.GetSourceResponse;
import org.elasticsearch.client.indices.*;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

/**
 * @author 邓力宾
 * @date 2020/12/25 11:32
 * @desc api 文档:https://www.elastic.co/guide/en/elasticsearch/client/java-rest/7.10/java-rest-high-supported-apis.html
 */
public class RestEsClient {
    private static final RequestOptions COMMON_OPTIONS;
    private static final String TOKEN="this is token";
    private RestHighLevelClient client;
    static {
        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
        builder.addHeader("Authorization", "Bearer " + TOKEN);
        builder.setHttpAsyncResponseConsumerFactory(
                new HttpAsyncResponseConsumerFactory
                        .HeapBufferedResponseConsumerFactory(30 * 1024 * 1024 * 1024));
        COMMON_OPTIONS = builder.build();
    }
    public RestEsClient(){
         client = new RestHighLevelClient(
                RestClient.builder(
                        new HttpHost("192.168.2.177", 9200, "http"),
                        new HttpHost("192.168.2.178", 9200, "http"),
                        new HttpHost("192.168.2.179", 9200, "http")));

    }
    public void close() throws IOException {
        client.close();
    }

    /**
     * 创建索引库
     * @param indexName
     * @param shardsNum
     * @param replicasNum
     * @throws IOException
     */
    public void createIndex(String indexName, int shardsNum, int replicasNum, Map<String, FieldType> mappingProperties, Consumer<String> consumer) throws IOException {
        CreateIndexRequest request = new CreateIndexRequest(indexName);
        request.settings(Settings.builder()
                .put("index.number_of_shards", shardsNum)
                .put("index.number_of_replicas", replicasNum)

        );


        Map<String, Object> mapping = new HashMap<>();
        mapping.put("properties", createMapping(mappingProperties));
        request.mapping(mapping);

        request.setTimeout(TimeValue.timeValueMinutes(2));
        request.setMasterTimeout(TimeValue.timeValueMinutes(1));

        request.waitForActiveShards(ActiveShardCount.from(2));
        request.waitForActiveShards(ActiveShardCount.DEFAULT);

        //同步执行
        //CreateIndexResponse createIndexResponse = client.indices().create(request, RequestOptions.DEFAULT);

        //异步执行
        client.indices().createAsync(request, RequestOptions.DEFAULT, new ActionListener<CreateIndexResponse>() {

            @Override
            public void onResponse(CreateIndexResponse createIndexResponse) {
                boolean acknowledged = createIndexResponse.isAcknowledged();
                boolean shardsAcknowledged = createIndexResponse.isShardsAcknowledged();
                consumer.accept("acknowledged:"+acknowledged+";shardsAcknowledged:"+shardsAcknowledged);
            }

            @Override
            public void onFailure(Exception e) {
                e.printStackTrace();
                consumer.accept("失败");
            }
        });
    }

    /**
     * 删除索引库
     * @param indexName
     */
    public void deleteIndex(String indexName, Consumer<String> consumer){
        DeleteIndexRequest request = new DeleteIndexRequest(indexName);
        request.timeout(TimeValue.timeValueMinutes(2));
        //request.timeout("2m");
        request.masterNodeTimeout(TimeValue.timeValueMinutes(1));
        //request.masterNodeTimeout("1m");
        request.indicesOptions(IndicesOptions.lenientExpandOpen());
        //同步请求
        //AcknowledgedResponse deleteIndexResponse = client.indices().delete(request, RequestOptions.DEFAULT);

        //异步
        client.indices().deleteAsync(request, RequestOptions.DEFAULT,  new ActionListener<AcknowledgedResponse>() {
            @Override
            public void onResponse(AcknowledgedResponse deleteIndexResponse) {
                boolean acknowledged = deleteIndexResponse.isAcknowledged();
                consumer.accept("acknowledged:"+acknowledged);
            }

            @Override
            public void onFailure(Exception e) {
               e.printStackTrace();
               consumer.accept("操作异常:"+e.getMessage());

            }
        });
    }


    /**
     * 是否存在
     * @param indexName
     */
    public boolean exist(String indexName) throws IOException {
        GetIndexRequest request = new GetIndexRequest(indexName);
        request.local(false); //Whether to return local information or retrieve the state from master node
        request.humanReadable(true);
        request.includeDefaults(false);

        //同步请求
        boolean exists = client.indices().exists(request, RequestOptions.DEFAULT);
        //异步请求
        //client.indices().existsAsync(request, RequestOptions.DEFAULT, listener);
        return exists;
    }

    /**
     * 打开索引（允许读写操作）
     * @param indexName
     * @param consumer
     */
    public void openIndex(String indexName, Consumer<String> consumer){
        OpenIndexRequest request = new OpenIndexRequest(indexName);
        request.timeout(TimeValue.timeValueMinutes(2));
        //request.timeout("2m");
        request.masterNodeTimeout(TimeValue.timeValueMinutes(1));
        //request.masterNodeTimeout("1m");

        request.waitForActiveShards(2);
        //request.waitForActiveShards(ActiveShardCount.DEFAULT);

        request.indicesOptions(IndicesOptions.strictExpandOpen());

        //同步执行
       // OpenIndexResponse openIndexResponse = client.indices().open(request, RequestOptions.DEFAULT);

        //异步执行
        client.indices().openAsync(request, RequestOptions.DEFAULT,  new ActionListener<OpenIndexResponse>() {
            @Override
            public void onResponse(OpenIndexResponse openIndexResponse) {
                consumer.accept("isAcknowledged:"+openIndexResponse.isAcknowledged()+
                        ";isShardsAcknowledged:"+openIndexResponse.isShardsAcknowledged());
            }

            @Override
            public void onFailure(Exception e) {
                e.printStackTrace();
                consumer.accept(e.getMessage());
            }
        });
    }

    /**
     * 关闭索引（禁止读写操作）
     * @param indexName
     * @param consumer
     */
    public void closeIndex(String indexName, Consumer<String>consumer){
        CloseIndexRequest request = new CloseIndexRequest("index");
        request.setTimeout(TimeValue.timeValueMinutes(2));
        request.setMasterTimeout(TimeValue.timeValueMinutes(1));
        request.indicesOptions(IndicesOptions.lenientExpandOpen());
        //同步请求
        //AcknowledgedResponse closeIndexResponse = client.indices().close(request, RequestOptions.DEFAULT);
        //异步请求
        client.indices().closeAsync(request, RequestOptions.DEFAULT, new ActionListener<CloseIndexResponse>() {
            @Override
            public void onResponse(CloseIndexResponse closeIndexResponse) {
                boolean acknowledged = closeIndexResponse.isAcknowledged();
                consumer.accept("acknowledged:"+acknowledged);
            }

            @Override
            public void onFailure(Exception e) {
                e.printStackTrace();
                consumer.accept(e.getMessage());
            }
        });
    }

    /**
     * 添加文档
     * @param indexName
     * @param id
     * @param doc
     * @param consumer
     */
    public void insertDoc(String indexName, String id, Map<String, Object> doc, Consumer<String> consumer){
        IndexRequest request = new IndexRequest(indexName);
        request.id(id).source(doc);
        //request.type("user"); //type已被弃用
        request.timeout(TimeValue.timeValueSeconds(30));
       // request.timeout("1s");

        request.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);


        request.opType(DocWriteRequest.OpType.CREATE);
        //request.opType("create");

        //request.setPipeline("pipeline");
        client.indexAsync(request, RequestOptions.DEFAULT, new ActionListener<IndexResponse>() {
            @Override
            public void onResponse(IndexResponse indexResponse) {
                StringBuilder sb = new StringBuilder();
                String index = indexResponse.getIndex();
                String id = indexResponse.getId();
                sb.append("index:"+index+"\n");
                sb.append("id:"+id+"\n");
                if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
                    sb.append("CREATED\n");
                } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
                    sb.append("UPDATED\n");
                }
                ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
                if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                    sb.append("处理成功的分片数不等于分片总数\n");
                }
                if (shardInfo.getFailed() > 0) {
                    sb.append("分片失败数:"+shardInfo.getFailed()+"\n");
                    sb.append("失败原因:\n");
                    for (ReplicationResponse.ShardInfo.Failure failure :
                            shardInfo.getFailures()) {
                        String reason = failure.reason();
                        sb.append("\t"+reason+"\n");
                    }
                }
                consumer.accept(sb.toString());
            }

            @Override
            public void onFailure(Exception e) {
                e.printStackTrace();
                consumer.accept(e.getMessage());
            }
        });

    }


    /**
     * 根据id查询
     * @param indexName
     * @param id
     * @return
     */
    public Map<String, Object>  getSource(String indexName, String id){
        GetSourceRequest getSourceRequest = new GetSourceRequest(indexName, id);
        try {
            GetSourceResponse response =
                    client.getSource(getSourceRequest, RequestOptions.DEFAULT);
            return response.getSource();
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
    }

    private  Map<String,Object> createMapping(Map<String, FieldType> mappingProperties){
        Map<String, Object> map = new HashMap<>();
        mappingProperties.forEach((fieldName, fieldType)->{
            Map<String, Object> m = new HashMap<>();
            m.put("type", fieldType.getType());
            m.put("store", fieldType.getStore());
            //使用ik分词器
            if(fieldType.getType().equals("text")){
                m.put("analyzer", "ik_max_word");
                //"search_analyzer": "ik_smart"
                m.put("search_analyzer", "ik_smart");
            }
            map.put(fieldName, m);
        });
        return map;
    }

}
